// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-07-16

package src

import (
	"context"
	"gitee.com/go-libs/process"
	"sync"
	"sync/atomic"
	"time"
)

// Crontab
// is a scheduler for crontab. It's worked as a provider for process.
type Crontab struct {
	mu      *sync.RWMutex
	name    string
	process process.Process
	workers map[string]*Worker

	scheduleFailed, scheduleSucceed uint64
	scheduleRunning                 int32
}

// NewCrontab
// creates a new crontab instance.
func NewCrontab(name string) *Crontab {
	o := &Crontab{mu: new(sync.RWMutex), name: name}
	o.process = process.New(o.name, o)
	o.workers = make(map[string]*Worker)
	return o
}

// Add
// adds a worker into crontab.
func (o *Crontab) Add(worker *Worker) error {
	o.mu.Lock()
	defer o.mu.Unlock()

	// Return error
	// if passed worker is nil.
	if worker == nil {
		return Error{
			Type:    "crontab",
			Message: "the worker you added can not be nil",
		}
	}

	// Return error
	// if job is nil or not allowed.
	if worker.job == nil {
		return Error{
			Type:    "worker",
			Message: "the job for worker not allowed",
		}
	}

	// Return error
	// if timed execution time not valid.
	var err error
	if worker.expression, err = worker.strategy.Parse(); err != nil {
		return Error{
			Type:    "strategy",
			Message: "invalid timed execution strategy for the worker",
			Value:   worker.strategy.String(),
		}
	}

	// Return error
	// if worker added already.
	if _, ok := o.workers[worker.class]; ok {
		return Error{
			Type:    "crontab",
			Message: "the worker added already",
			Value:   worker.class,
		}
	}

	// Update
	// workers mapper.
	o.workers[worker.class] = worker
	return nil
}

// Del
// deletes a worker from crontab.
func (o *Crontab) Del(worker *Worker) {
	if worker != nil {
		o.DelByClass(worker.class)
	}
}

// DelByClass
// deletes a worker from crontab by class name.
func (o *Crontab) DelByClass(class string) {
	o.mu.Lock()
	defer o.mu.Unlock()

	// Remove
	// from mapping if exists.
	if _, ok := o.workers[class]; ok {
		delete(o.workers, class)
	}
}

// GetName
// returns the crontab name.
func (o *Crontab) GetName() string {
	return o.name
}

// GetWorker
// returns the worker with given class.
func (o *Crontab) GetWorker(class string) *Worker {
	o.mu.RLock()
	defer o.mu.RUnlock()
	if worker, ok := o.workers[class]; ok {
		return worker
	}
	return nil
}

// GetWorkers
// returns the added worker mapping.
func (o *Crontab) GetWorkers() map[string]*Worker {
	o.mu.RLock()
	defer o.mu.RUnlock()
	return o.workers
}

// Process
// returns a process of the crontab.
func (o *Crontab) Process() process.Process {
	return o.process
}

// Start
// starts the crontab service by call process.
func (o *Crontab) Start(ctx context.Context) (err error) {
	return o.process.Start(ctx)
}

// Started
// return true if process started.
func (o *Crontab) Started() bool {
	return o.process.Started()
}

// Stop
// stops the crontab service by call process.
func (o *Crontab) Stop() {
	o.process.Stop()
}

// Stopped
// return true if process stopped.
func (o *Crontab) Stopped() bool {
	return o.process.Stopped()
}

// +---------------------------------------------------------------------------+
// | Implemented on process                                                    |
// +---------------------------------------------------------------------------+

// After
// wait all running scheduler done before stopped. It's called by process, do not
// call it directly.
func (o *Crontab) After(_ context.Context) (err error) {
	for {
		if atomic.LoadInt32(&o.scheduleRunning) == 0 {
			return
		}
		time.Sleep(time.Millisecond * 10)
	}
}

// Run
// crontab service. It's called by process, do not call it directly.
func (o *Crontab) Run(ctx context.Context) (err error) {
	// Creates
	// a ticker to schedule.
	tick := time.NewTicker(time.Second)

	// Range
	// ticker and context channel.
	for {
		select {
		case tm := <-tick.C:
			go o.schedule(ctx, tm)
		case <-ctx.Done():
			tick.Stop()
			tick = nil
			return
		}
	}
}

// +---------------------------------------------------------------------------+
// | Access methods                                                            |
// +---------------------------------------------------------------------------+

func (o *Crontab) schedule(ctx context.Context, tm time.Time) {
	// Iterate
	// added workers in crontab.
	for _, x := range func() map[string]*Worker {
		o.mu.RLock()
		defer o.mu.RUnlock()
		return o.workers
	}() {
		// Increment running statistic.
		atomic.AddInt32(&o.scheduleRunning, 1)
		// Schedule in a goroutine.
		go func(worker *Worker) {
			// Decrement
			// when done.
			defer atomic.AddInt32(&o.scheduleRunning, -1)
			// Run worker.
			if do, err := worker.run(ctx, tm); do {
				// Increment
				// result statistic.
				if err == nil {
					// Succeed
					// to run.
					atomic.AddUint64(&o.scheduleSucceed, 1)
				} else {
					// Failed
					// to run.
					atomic.AddUint64(&o.scheduleFailed, 1)
				}
			}
		}(x)
	}
}
